package com.atguigu.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.uitl.DruidDSUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

/**
 * @author hjy
 * @create 2023/3/10 10:54
 */
public class DimJdbcFunction extends RichSinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取连接
        DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
        //拼接sql
        String sql=genUpsertSQL(value.getString("sink_table"),value.getJSONObject("data"));
        System.out.println("upsert语句:"+sql);
        //插入数据
        PreparedStatement preparedStatement = phoenixConn.prepareStatement(sql);
        preparedStatement.execute();
        phoenixConn.commit();
        //关闭连接
        phoenixConn.close();
    }
    //upsert into db.t (id,name,url) values('1001','zs','atguigu')
    private String genUpsertSQL(String sink_table, JSONObject data) {
        Set<String> keys = data.keySet();
        Collection<Object> values = data.values();
        return "upsert into "
                + GmallConfig.HBASE_SCHEMA
                +"."
                +sink_table
                +" ( "
                + StringUtils.join(keys,",")
                +") values ('"
                +StringUtils.join(values,"','")
                +"')";
    }
}
